package spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.tree.RandomForest;
import org.apache.spark.mllib.tree.model.RandomForestModel;
import org.apache.spark.mllib.util.MLUtils;
import scala.Tuple2;

import java.util.HashMap;

/**
 * 作者: LDL
 * 功能说明:
 * 创建日期: 2015/7/3 10:11
 */
public class RandomForests {
    public static void main(String[] args) {
        System.setProperty("hadoop.home.dir", "F:\\tools\\hadoop-common-2.2.0-bin-master");
        SparkConf conf = new SparkConf().setMaster("local").setAppName("JavaRandomForestClassification");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        //jsc.setLogLevel("OFF");
        JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotrans\\libsvm.txt").toJavaRDD();
        JavaRDD<LabeledPoint> test = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotest\\libsvm.txt").toJavaRDD();
        // Split the data into training and test sets (30% held out for testing)
        /*JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.7, 0.3});
        JavaRDD<LabeledPoint> trainingData = splits[0];
        JavaRDD<LabeledPoint> testData = splits[1];*/

        // Train a RandomForest model.
        //  Empty categoricalFeaturesInfo indicates all features are continuous.
        Integer numClasses = 2;
        HashMap<Integer, Integer> categoricalFeaturesInfo = new HashMap<Integer, Integer>();
        Integer numTrees = 3; // Use more in practice.
        String featureSubsetStrategy = "auto"; // Let the algorithm choose.
        String impurity = "gini";
        Integer maxDepth = 15;
        Integer maxBins = 32;
        Integer seed = 12345;

        final RandomForestModel model = RandomForest.trainClassifier(data, numClasses,
                categoricalFeaturesInfo, numTrees, featureSubsetStrategy, impurity, maxDepth, maxBins,
                seed);

        // Evaluate model on test instances and compute test error
        JavaPairRDD<Double, Double> predictionAndLabel =
                test.mapToPair(p -> new Tuple2<>(model.predict(p.features()), p.label()));
        Double testErr =
                1.0 * predictionAndLabel.filter(pl -> !pl._1().equals(pl._2())).count() / test.count();
        System.out.println("Test Error: " + testErr);
        //System.out.println("Learned classification forest model:\n" + model.toDebugString());
    }
}
